package com.bw.app.functions;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;


/**
 * {
 * database:xxx
 * table:xxx
 * before:{"name":xxx}
 * after:{"age":xxx}
 * op:
 * ts:
 * }
 */
public class CustomDebezium implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        // 1.创建JsonObject对象
        JSONObject result = new JSONObject();
        // 2. 取出库名和表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        String database = fields[1];
        String tableName = fields[2];
        // 3.取出before数据
        Struct value = (Struct) sourceRecord.value();
        Struct before = value.getStruct("before");
        JSONObject beforeJsonObject = new JSONObject();
        if (before != null) {
            Schema beforeSchema = before.schema();
            // 取出所有字段
            List<Field> beforeFields = beforeSchema.fields();
            for (int i = 0; i < beforeFields.size(); i++) {
                // 取出每个属性
                Field beforeField = beforeFields.get(i);
                // 根据属性取出每个属性的值
                Object beforeValue = before.get(beforeField);
                beforeJsonObject.put(beforeField.name(),beforeValue);
            }
        }

        // 4.取出after数据

        Struct after = value.getStruct("after");
        JSONObject afterJsonObject = new JSONObject();
        if (after != null) {
            Schema afterSchema = after.schema();
            // 取出所有字段
            List<Field> afterFields = afterSchema.fields();
            for (int i = 0; i < afterFields.size(); i++) {
                // 取出每个属性
                Field afterField = afterFields.get(i);
                // 根据属性取出每个属性的值
                Object afterValue = after.get(afterField);
                afterJsonObject.put(afterField.name(),afterValue);
            }
        }
        // 5.取出操作类型
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String op= operation.toString().toLowerCase();
        if ("create".equals(op)){
            op = "insert";
        }
        // 6.封装成json
        result.put("database", database);
        result.put("tableName", tableName);
        result.put("before",  beforeJsonObject);
        result.put("after", afterJsonObject);
        result.put("type", op);
        // 收集数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
